/*
 * Version 0.2.1
 *
 * Copyright (c) 2010 Guet
 *
 * All rights reserved.
 *
 * Developed by: XGuru
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <time.h>
#include <google/tcmalloc.h>
#include "flycache.h"

static struct settings g_setting;
static struct hashtable *g_hashtable;
/*static struct already_read_data g_data;*/
static struct fly_status g_status;

int setnonblocking(int sockfd)
{
    if (fcntl(sockfd,F_SETFL,fcntl(sockfd,F_GETFD,0)|O_NONBLOCK) == -1)
    {
        return -1;
    }
    return 0;
}
void exit_program(int sig) 
{
	printf("\nCatched signal: %d ... \n", sig);
	close(g_status.sockfd); 
	signal(SIGINT, SIG_DFL);
	exit(EXIT_SUCCESS); 
}
void chomp(char *str)
{
    if ( str[ strlen(str)-1 ] == '\n')
    {
        if (str[ strlen(str)-2 ] == '\r')
            str[ strlen(str)-2] = '\0';
        else
            str[ strlen(str)-1] = '\0';
    }
}
int send_info(int sockfd,char *info)
{
    if (write(sockfd,info,strlen(info)) == -1)
    {
        perror("write error:");
        return -1;
    }
    else
        return 0;
}

void init_setting()
{
    g_setting.port       = DEFAULT_PORT;
    g_setting.is_deamon  = 0;
    g_setting.max_client = DEFAULT_MAX_CLIENT;
    g_setting.verbose    = 0;
    g_setting.max_memory = DEFAULT_MAX_MEMORY;
}
void init_status()
{
    /*sprintf(g_status.version,"%s",VERSION);*/
    g_status.pid = getpid();
    time(&g_status.start_time);
    g_status.curr_connections = 0;
    g_status.curr_items       = 0;
    g_status.uptime           = 0;
    g_status.limit_maxbytes   = 0;
    g_status.total_items      = 0;
}
void init_socket()
{
    int ret,connfd,listenfd,epollfd,nfds,i,curfds;
    int continue_read = 0;
    socklen_t chilen;
    struct sockaddr_in server_addr,client_addr;
    struct epoll_event ev;
    struct epoll_event events[MAX_EPOLL_SIZE];
    memset(&server_addr,0,sizeof(server_addr));
    memset(&client_addr,0,sizeof(client_addr));
    if ( (listenfd = socket(AF_INET,SOCK_STREAM,0)) < 0)
    {
        printf("socket error!\n");
        exit(EXIT_FAILURE);
    }
	g_status.sockfd = listenfd; 
    server_addr.sin_family      = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port        = htons(g_setting.port);

    if ( bind(listenfd,(struct sockaddr *)&server_addr,sizeof(server_addr)) < 0 )
    {
        printf("bind error!\n");
        exit(EXIT_FAILURE);
    }

    if ( listen(listenfd,128) < 0 )
    {
        printf("listen error!\n");
        exit(EXIT_FAILURE);
    }

    epollfd = epoll_create(MAX_EPOLL_SIZE);
    if (epollfd == -1)
    {
        printf("epoll create error!");
        exit(EXIT_FAILURE);
    }
    ev.events  = EPOLLIN | EPOLLET;
    ev.data.fd = listenfd;
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev) < 0)
    {
        printf("epoll_ctl error: listen_sock");
		exit(EXIT_FAILURE);
    }
    curfds = 1;
	signal(SIGINT, exit_program); 
    while (1)
    {
        nfds = epoll_wait(epollfd, events, curfds, -1);
        if (nfds == -1)
        {
            perror("epoll_wait");
            exit(EXIT_FAILURE);
        }
        for (i = 0; i < nfds ; i++)
        {
            if (events[i].data.fd == listenfd)
            {

                chilen = sizeof(client_addr);
                if (( connfd = accept(listenfd,(struct sockaddr *)&client_addr,&chilen)) <0)
                {
                    printf("accept error!\n");
                    continue;
                }
				if (g_status.curr_connections  >= g_setting.max_client)
					 close(connfd); 
				else
				{	
					g_status.curr_connections++;
					setnonblocking(connfd);
					ev.events = EPOLLIN | EPOLLET;
					ev.data.fd = connfd;
					if (epoll_ctl(epollfd,EPOLL_CTL_ADD,connfd,&ev)== -1)
					{
						perror("epoll_ctl connfd error!\n");
						exit(1);
					}
					curfds++;
				}
            }
            else
            {
                ret = dispose_conn(events[i].data.fd,continue_read);
                if (ret == RET_EXIT )
                {
                    epoll_ctl(epollfd,EPOLL_CTL_DEL,events[i].data.fd,&ev);
                    curfds--;
                    g_status.curr_connections--;
                    close(events[i].data.fd);
                    break;
                }
                else if ( ret == RET_CONTINUE )
                    continue_read = 1;
                else if (ret == RET_SUCCESS)
                    continue_read = 0;
            }
        }
    }
    close(listenfd);
}

void print_usage(char *argv)
{
    printf("\n*flycache v%s* \n\n",VERSION);
    printf("usage:%s [-p num] [-c num] [-d] [-v] [-h] \n",argv);
    printf(" -p [num]: set listening port number \n");
    printf(" -c [num]: max clients connection numbers \n");
    printf(" -d : runnig as daemon \n");
    printf(" -v : verbose information \n");
    printf(" -V : print version\n");
    printf(" -h : help \n\n");
}
int check_param(char *token[],int n)
{
	int i; 
	for(i =0; i<n; i++)
	{
		if (token[i]  == NULL)
			return -1; 
	}
	return 0; 
}
int dispose_conn(int sockfd,int continue_read)
{
    char *request_buf;
    char *data;
    char *key;
    char buf[BUFF_SIZE];
    char *token[BUFF_SIZE];
    char *method;
    int i = 0, j = 0;
    FILE *fd;
    char request[BUFF_SIZE];
    fd = fdopen(sockfd,"r");
    if (fgets(request,BUFF_SIZE,fd) == NULL)
    {
        if ( g_setting.verbose > 0)
            perror("fget message");
        return RET_EXIT;
    }
    request_buf = request;
    chomp(request_buf);
    while ((token[i] = strsep(&request_buf, " ")) != NULL)
    {
        /*printf("%s\n",token[i]);*/
        i++;
    }
    method = token[0];
    if (STRING_EQU(method,"set"))
    {
		if (check_param(token, 6)  == -1)
			send_info(sockfd, "PARAMETER ERROR\r\n"); 
		else
		{
			int length;
			int left_length; 
			key  = (char *)malloc( strlen(token[1]) + 1);
			strcpy(key,token[1]);
			length  = atoi(token[4]);
			data    = (char *)malloc( length + 1);
			strncpy(data,token[5],length+1);
			left_length = length - strlen(token[5]); 
			j = 6;
			//如果还有数据没有读完，继续读 
			while( (token[j]  != NULL)  &&  (left_length  > 0))
			{
				strncat(data," ", 1); 
				left_length -= 1;
				strncat(data, token[j], left_length); 
				left_length -= strlen(token[j]);
				if((token[j+1]  != NULL)  &&  (left_length < strlen(token[j+1])))
				{
					strncat(data, token[j+1], left_length); 
					break; 
				}
				printf("%d", j); 
				j++; 
			}
			store_item(key,token[2],token[3],token[4],data);
			send_info(sockfd,"STORE\r\n");
			/* key     = token[1]
			 * flag    = token[2];
			 * exptime = token[3];
			 * length  = token[4];
			 * data    = token[5];*/
		}
    }
    else if (STRING_EQU(method,"get"))
    {
		if (check_param(token, 2)  == -1)
			send_info(sockfd, "PARAMETER ERROR\r\n"); 
		else
		{
			struct hashentry *entry;
			if ( g_setting.verbose > 0)
				printf("token[1]%s\n",token[1]);
			if ((entry = get_item(token[1])) == NULL)
			{
				send_info(sockfd,"NOT FOUND\r\n");
			}
			else
			{
				memset(buf, 0,BUFF_SIZE);
				sprintf(buf,"VALUE %s %d %d\r\n%s\r\nEND\r\n",entry->key,entry->flag,entry->length,entry->value);
				send_info(sockfd,buf);
			}
		}
    }
    else if (STRING_EQU(method,"delete"))
    {
		if (check_param(token, 2)  == -1)
			send_info(sockfd, "PARAMETER ERROR\r\n"); 
		else
		{
			if (delete_item(token[1]) == 0)
			{
				send_info(sockfd,"DELETED\r\n");
			}
			else
				send_info(sockfd,"NOT_FOUND\r\n");
		}
    }
    else if (STRING_EQU(method,"stats"))
    {
        send_stats(sockfd);
    }
    else if (STRING_EQU(method,"flush_all"))
    {
        flush_all();
    }
    else if (STRING_EQU(method,"quit"))
    {
        send_info(sockfd,"QUIT\r\n");
        close(sockfd);
        return RET_EXIT;
    }
    else
        send_info(sockfd,"ERROR\r\n");
    return RET_SUCCESS;
}
int send_stats(int sockfd)
{
    time(&g_status.time);
    char buf[BUFF_SIZE];
    g_status.uptime = g_status.time - g_status.start_time;
    sprintf(buf, "STAT version 0.21\r\nSTAT pid %d\r\nSTAT uptime %ld\r\nSTAT time %ld\r\nSTAT mem_total %d\r\nSTAT item_total %d\r\nSTAT curr_item %d\r\nSTAT curr_connections %d\r\nEND\r\n",
            g_status.pid,
            g_status.uptime,
            g_status.time,
            g_status.limit_maxbytes,
            g_status.total_items,
            g_status.curr_items,
            g_status.curr_connections);
    send_info(sockfd,buf);
    return 0;
}
int store_item(char *key,char *flag,char *exptime,char *length,char *value)
{
    int item_flag,item_length;
    unsigned int item_exptime;
    if ( g_setting.verbose > 0)
        printf("####Store item\n#key:%s\n#value:%s\n",key,value);
    item_flag = atoi(flag);
    item_length = atoi(length);
    item_exptime = atoi(exptime);
    if ( hash_insert(g_hashtable,key,item_flag,item_exptime,item_length,value) == -1)
        return -1;
    g_status.curr_items++;
    return 0;
}

struct hashentry *get_item(char *key)
{
    struct hashentry *entry;
    time_t currtime;
    time(&currtime);
    entry =    hash_lookup(g_hashtable,key);
    if ( g_setting.verbose > 0)
    {
        printf("####Get item\n####key:%s\n",key);
        if ( entry != NULL)
        {
            printf("init_time+exptime = %ld\n",entry->init_time + entry->exptime);
            printf("currtime: %ld\n",currtime);
        }
    }
    if ( (entry != NULL) && (entry->init_time + entry->exptime < currtime) && (entry->exptime != 0) )
    {
        delete_item(key);
        return NULL;
    }
    return entry;
}
int delete_item(char *key)
{
    if ( g_setting.verbose > 0)
        printf("####Delete item\n####key:%s\n",key);
    if (hash_remove(g_hashtable,key) == -1)
        return -1;
    else
    {
        g_status.curr_items--;
        return 0;
    }
}

void flush_all()
{
    hash_free(g_hashtable);
    g_hashtable = hash_init(MAX_HASH_SIZE);
}
int main(int argc, char *argv[])
{
    int opt;
    g_hashtable = hash_init(MAX_HASH_SIZE);
    init_setting();
    while ((opt = getopt(argc,argv,
                         "p:"/* Port g_setting */
                         "c:"/* Maximum client numbers*/
                         "m:"
                         "d" /* Daemon or not */
                         "h" /* Show manual*/
                         "v" /* verbose information */
                         "V" /* Show version */
                        )) != -1)
        switch (opt)
        {
        case 'p':
            g_setting.port = atoi(optarg);
            printf("port %d\n",g_setting.port);
            break;
        case 'c':
            if (atoi(optarg) < MAX_EPOLL_SIZE)
                g_setting.max_client = atoi(optarg);
            else
            {
                printf("max client can't beyond %d\n",MAX_EPOLL_SIZE);
                exit(EXIT_FAILURE);
            }
            break;
        case 'm':
            if (atoi(optarg) < MAX_MEMORY)
                g_setting.max_memory = atoi(optarg);
            else
            {
                printf("max memory can't beyond %d\n",MAX_MEMORY);
                exit(EXIT_FAILURE);
            }
            break;
        case 'd':
            printf("daemon\n");
            g_setting.is_deamon = 1;
            break;
        case 'h':
            print_usage(argv[0]);
            exit(EXIT_SUCCESS);
        case 'v':
            g_setting.verbose = 1;
            break;
        case 'V':
            printf("flycache version is %s\n",VERSION);
            exit(EXIT_SUCCESS);
        }
    if (g_setting.is_deamon == 1)
    {
        if (daemon(0,0) < 0)
        {
            perror("daemon");
            exit(EXIT_FAILURE);
        }
    }
    init_status();
    init_socket();
    return 0;
}
